package com.wlz.rocketmq.simple;

import com.wlz.rocketmq.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * @author wlz
 * @date 2022-07-01  10:29 下午
 */
@Slf4j
public class LitePullConsumerAssign {

    public static volatile  boolean running = true;

    public static void main(String[] args) throws Exception{

        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("please_rename_unique_group_name_5");

        consumer.setNamesrvAddr(Constants.NAMESVADDR);

        consumer.setAutoCommit(false);

        consumer.start();

        Collection<MessageQueue> topicTest = consumer.fetchMessageQueues("TopicTest");

        List<MessageQueue> messageQueues = new ArrayList<>(topicTest);
        List<MessageQueue> assignList = new ArrayList<>();
        for (int i = 0; i < messageQueues.size()/2; i++) {
            assignList.add(messageQueues.get(i));
        }

        consumer.assign(assignList);
        consumer.seek(assignList.get(0), 10);

        try {
            while (running) {
                List<MessageExt> poll = consumer.poll();
                log.info("poll : {}", poll);
                consumer.commitSync();
            }
        } catch (Exception e) {
            log.error("LitePullConsumerAssign error: {}", e);
        }


    }
}
